import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._

// 定义一个名为UpdateStudents的对象
object UpdateStudents {
  def main(args: Array[String]): Unit = {
    // 创建一个SparkSession对象，设置应用名称为"UpdateStudents"，并指定在本地[*]上运行
    val spark = SparkSession.builder.appName("UpdateStudents").master("local[*]").getOrCreate()

    // 读取MySQL数据库中的表"students"数据到DataFrame
    val studentsDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/sparksql_practice")  // 数据库连接URL
      .option("dbtable", "students")  // 表名
      .option("user", "root")  // 用户名
      .option("password", "123456")  // 密码
      .option("driver", "com.mysql.jdbc.Driver")  // MySQL驱动程序
      .load()

    // 更新特定学生的专业为“Data Science”
    val updatedDF = studentsDF.withColumn("major", when(col("student_id") === 1, "Data Science").otherwise(col("major")))

    // 增加所有学生年龄1岁
    val updatedAgeDF = updatedDF.withColumn("age", col("age") + 1)

    // 打印更新后的数据
    updatedAgeDF.show()

    // 将更新后的数据重新写入数据库
    updatedAgeDF.write
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/sparksql_practice")  // 数据库连接URL
      .option("dbtable", "students1")  // 表名
      .option("user", "root")  // 用户名
      .option("password", "123456")  // 密码
      .option("driver", "com.mysql.cj.jdbc.Driver")  // MySQL驱动程序（更新为com.mysql.cj.jdbc.Driver）
      .mode("overwrite")  // 覆盖已存在的表数据
      .save()

    // 展示更新后的数据
    updatedAgeDF.show()

    spark.stop()  // 停止SparkSession
  }
}
